import java.text.SimpleDateFormat
import java.util
import java.util.Date

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{CouponAlertInfo, EventLog}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}

import scala.util.control.Breaks._

object AlertApp {
  def main(args: Array[String]): Unit = {
    //1.创建SparkConf
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AlertApp")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf,Seconds(5))

    //3.从kafka消费数据
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_EVENT,ssc)

    //4.将数据转为样例类,并返回kv类型
    val sdf: SimpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH")
    val eventLogDStream = kafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val eventLog: EventLog = JSON.parseObject(record.value(), classOf[EventLog])
        //补全字段
        val times: String = sdf.format(new Date(eventLog.ts))
        eventLog.logDate = times.split(" ")(0)
        eventLog.logHour = times.split(" ")(1)

        (eventLog.mid,eventLog)
      })
    })

    //5.开窗，开启一个5min的滑动窗口
    //如果窗口只传一个参数，则默认滑动步长为当前批次时间
    val windowDStream = eventLogDStream.window(Minutes(5))

    //6.分组聚合（mid）
    //迭代器中放的数据就是5分钟内相同mid的数据
    val midToIterEventLogDStream: DStream[(String, Iterable[EventLog])] = windowDStream.groupByKey()

    //7.根据条件筛选数据,并生成疑似预警日志
    /**
      * 三次及以上用不同账号登录并领取优惠劵，（判断用户行为是否有领优惠券行为，如果领优惠券，则要对这些用户做去重（set幂等性来去重），去重后判断不同        的用户数是否大于等于三）
      * 并且过程中没有浏览商品。（判断用户是否有浏览商品行为，一旦发现用户5分钟内有浏览商品行为，则就不符合预警要求）
      */
    val boolToCouponAlertInfoDStream: DStream[(Boolean, CouponAlertInfo)] = midToIterEventLogDStream.mapPartitions(partition => {
      partition.map { case (mid, iter) =>
        //创建set集合用来存放用户id
        val uids: util.HashSet[String] = new util.HashSet[String]()

        //创建set集合用来存放领优惠券所涉及的商品
        val itemIds: util.HashSet[String] = new util.HashSet[String]()

        //创建List集合用来存放用户所涉及的事件
        val events: util.ArrayList[String] = new util.ArrayList[String]()


        //创建一个标志位，用来判断用户是否有浏览商品行为
//        var bool: Boolean = true

        //遍历迭代器，获取每一个数据
        breakable {
          iter.foreach(log => {
            //添加用户涉及行为
            events.add(log.evid)
            //判断用户是否有浏览商品行为
            if ("clickItem".equals(log.evid)) {
              //一旦有浏览商品行为，则证明此5分钟内的数据不符合预警要求，则跳出循环
              uids.clear()
              break()
            } else if ("coupon".equals(log.evid)) {
              //用户没有浏览商品，但是领优惠券
              //需要把符合要求的用户放入set集合中去重，后续通过集合的长度判断是否符合预警需求
              uids.add(log.uid)
              //添加涉及商品的id
              itemIds.add(log.itemid)
            }
          })
        }
        //返回一个疑似预警日志 k(放的是用来判断是否是预警日志的条件),v（可能是预警日志的数据）
        (uids.size() >= 3, CouponAlertInfo(mid, uids, itemIds, events, System.currentTimeMillis()))
      }
    })
    

    //8.生成预警日志
    val couponAlertInfoDStream: DStream[CouponAlertInfo] = boolToCouponAlertInfoDStream.filter(_._1).map(_._2)
    couponAlertInfoDStream.print()

    //9.将预警日志写入ES
    couponAlertInfoDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partiton=>{
        val list: List[(String, CouponAlertInfo)] = partiton.toList.map(log => {
          //将数据封装成kv类型  k：指的是doc_id  v：指的是存放到es的数据
          (log.mid + log.ts / 1000 / 60, log)
        })
        MyEsUtil.insertBulk("1027",list)
      })
    })

    //10.启动任务并阻塞
    ssc.start()
    ssc.awaitTermination()
  }

}
